package com.atguigu.PracticalProject.app.dwd.db;

import com.atguigu.PracticalProject.app.BaseSqlApp;
import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class DwdTradePayDetailSuc extends BaseSqlApp {

    public static void main(String[] args) {
        new DwdTradePayDetailSuc().init(
                3008,
                2,
                "DwdTradePayDetailSuc"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {


        //1:trade_order_detail
        tEnv.executeSql("create table dwd_trade_order_detail( " +
                "id string, " +
                "course_id string, " +
                "course_name string, " +
                "order_id string, " +
                "user_id string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "create_time string, " +
                "update_time string, " +
                "province_id string, " +
                "session_id string, " +
                "order_status string, " +
                "ts bigint ," +
                "row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSource(com.atguigu.PracticalProject.common.Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "DwdTradePayDetailSuc"));

        //2。payment_info

        readOdsDb(tEnv,"Dwd_08_DwdTradePayDetailSuc");

        Table paymentInfo = tEnv.sqlQuery("select " +
                "data['order_id'] order_id, " +
                "data['payment_type'] payment_type, " +
                "data['callback_time'] callback_time, " +
                "ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='payment_info' " +
                "and `data`['payment_status']='1602'");

        tEnv.createTemporaryView("payment_info",paymentInfo);

        //3join
        Table result = tEnv.sqlQuery("select " +
                "od.id order_detail_id, " +
                "od.user_id user_id, " +
                "od.course_id, " +
                "od.course_name, " +
                "od.order_id, " +
                "od.origin_amount, " +
                "od.coupon_reduce, " +
                "od.final_amount, " +
                "od.create_time, " +
                "od.update_time, " +
                "od.province_id, " +
                "od.session_id, " +
                "od.order_status, " +
                "pi.payment_type, " +
                "pi.callback_time, " +
                "pi.ts, " +
                "od.row_op_ts row_op_ts " +
                "from payment_info pi " +
                "join dwd_trade_order_detail od on pi.order_id=od.order_id ");

        //4:写出到kafka
        tEnv.executeSql("create table dwd_trade_pay_detail_suc( " +
                "order_detail_id string, " +
                "user_id string, " +
                "course_id string, " +
                "course_name string, " +
                "order_id string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "create_time string, " +
                "update_time string, " +
                "province_id string, " +
                "session_id string, " +
                "order_status string, " +
                "payment_type string, " +
                "callback_time string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")"+ SQLUtil.getKafkaSink(com.atguigu.PracticalProject.common.Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC)
        );


        result
                //.execute().print();
                .executeInsert("dwd_trade_pay_detail_suc");



    }
}
